package spring.boot.rocketmq.send.controller;

import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import spring.boot.rocketmq.send.service.TransactionListenerImpl;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.concurrent.*;
import java.util.List;



@RestController
@RequestMapping("/rmq")
@Log4j2
public class ProducerController {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    private TransactionMQProducer transactionMQProducer;

    /**
     * 同步发送
     * @param topic
     * @param tags
     * @param keys
     * @param info
     * @throws Exception
     */
    @GetMapping("/sync")
    public void sync(String topic, String tags, String keys,String info) throws Exception {
        Message msg = new Message(topic,tags,keys,info.getBytes(RemotingHelper.DEFAULT_CHARSET));
        transactionMQProducer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                logger.info("传输成功{}",sendResult.getMessageQueue().toString());
            }
            @Override
            public void onException(Throwable e) {
                logger.error("传输失败", e);
            }
        });
    }
    /**
     * 异步发送
     * @param topic
     * @param tags
     * @param keys
     * @param info
     * @throws Exception
     */
    @GetMapping("/async")
    public void async(String topic, String tags, String keys,String info) throws Exception {
        Message message = new Message(topic, tags, keys, info.getBytes(RemotingHelper.DEFAULT_CHARSET));
        transactionMQProducer.setRetryTimesWhenSendAsyncFailed(0);
        transactionMQProducer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                logger.info("传输成功{}",sendResult.getMessageQueue().toString());
            }
            @Override
            public void onException(Throwable e) {
                logger.error("传输失败", e);
            }
        });
    }

    /**
     * 队列发送
     * @param topic
     * @param tags
     * @param keys
     * @param info
     * @throws Exception
     */
    @GetMapping("/queue")
    public void queue(String topic, String tags, String keys,String info) throws Exception {
        for (int i = 0; i < 5; i++) {
            Message msg = new Message(topic,tags,keys,info.getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = transactionMQProducer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Integer id = (Integer) arg;
                    int index = id % mqs.size();
                    return mqs.get(index);
                }
            }, 0);
            logger.info("传输成功{}",sendResult.getMessageQueue().toString());
        }
    }

    /**
     * 事物
     * @param topic
     * @param tags
     * @param keys
     * @param info
     * @throws Exception
     */
    @GetMapping("/transaction")
    public void transaction(String topic, String tags, String keys,String info) throws Exception {
        TransactionListener transactionListener = new TransactionListenerImpl();
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });
        transactionMQProducer.setExecutorService(executorService);
        transactionMQProducer.setTransactionListener(transactionListener);
        try {
            Message msg = new Message(topic,tags,keys,info.getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult sendResult = transactionMQProducer.sendMessageInTransaction(msg, null);
            System.out.printf("%s%n", sendResult);
        } catch (MQClientException | UnsupportedEncodingException e) {
            System.out.println("====error=="+e.getMessage());
        }
    }



    /**
     * One-way transmission is used for cases requiring moderate reliability, such as log collection
     * @param topic
     * @param tags
     * @param keys
     * @param info
     * @throws Exception
     */
    @GetMapping("/oneWay")
    public void oneWay(String topic, String tags, String keys,String info) throws Exception {
        Message msg = new Message(topic,tags,keys,info.getBytes(RemotingHelper.DEFAULT_CHARSET));
        transactionMQProducer.sendOneway(msg);
    }

    /**
     * //默认级别：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，也就是说delayLevel = 3代表延迟10秒后重投递
     * @param topic
     * @param tags
     * @param keys
     * @param info
     * @throws Exception
     */
    @GetMapping("/scheduled")
    public void scheduled(String topic, String tags, String keys,String info) throws Exception {
        Message msg = new Message(topic,tags,keys,info.getBytes(RemotingHelper.DEFAULT_CHARSET));

        msg.setDelayTimeLevel(3);
        transactionMQProducer.send(msg);
    }

    /**
     * 不支持定时
     * 一批消息的大小 不应超过1M
     * @param topic
     * @param tags
     * @param keys
     * @param info
     * @throws Exception
     */
    @GetMapping("/batch")
    public void batch(String topic, String tags, String keys,String info) throws Exception {
        List<Message> messages = new ArrayList<>();
        messages.add(new Message(topic, tags, "OrderID001", "Hello world 0".getBytes()));
        messages.add(new Message(topic, tags, "OrderID002", "Hello world 1".getBytes()));
        messages.add(new Message(topic, tags, "OrderID003", "Hello world 2".getBytes()));
        try {
            transactionMQProducer.send(messages);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
